package com.behavioranalysis.flinkprogram.flink.ad;

import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.LinkedList;
import java.util.List;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午8:09
 * updated by yang on 20-2-14 下午8:09
 */
public class AssignerWithPeriodicWatermarksImpl implements
        AssignerWithPeriodicWatermarks<Tuple6<Long, String, String, String, String, String>> {
    private static final long serialVersionUID = 1L;
    // 1.分配时间戳，并生成水印
    private long currentMaxTimestamp = 0L;
    private long maxOutOfOrderness = 5000L;//最大乱序时间为5s

    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }

    @Override
    public long extractTimestamp(Tuple6<Long, String, String, String, String, String> element, long previousElementTimestamp) {
        long timestamp = element.f0;
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);

        List<Integer> list = new LinkedList<>();
        ((LinkedList<Integer>) list).remove();
        list.add()

        /**
         *
         */
        System.out.println("timestamp:"
                + element.f0 + "|" + format.format(element.f0) + ","
                + currentMaxTimestamp + "|" + format.format(currentMaxTimestamp) + ","
                + getCurrentWatermark().toString());
        return timestamp;
    }
}
